package org.apache.rocketmq.flink.common.serialization;

//import org.apache.flink.api.connector.source.SourceOutput;

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import java.io.Serializable;

/**
 * @program: fire
 * @description: 兼容 DeserializationSchema 序列化的多种实现，
 * 如 DebeziumJsonDeserializationSchema 的 debezium-json 序列化已经不再支持 deserialize(byte[]) 方法，
 * 只支持序列化方法：deserialize(byte[], Collector)
 * @author: TCM
 * @create: 2023-11-25 18:53
 **/
public class JuszSourceContext<T> implements SourceFunction.SourceContext<T>, Collector<T>, Serializable {
    private final SourceFunction.SourceContext<T> output;

    public JuszSourceContext(SourceFunction.SourceContext<T> output) {
        this.output = output;
    }

    public void collect(T record) {
        this.output.collect(record);
    }

    @Override
    public void collectWithTimestamp(T t, long l) {
        this.output.collectWithTimestamp(t, l);
    }

    @Override
    public void emitWatermark(Watermark watermark) {
        this.output.emitWatermark(watermark);
    }

    @Override
    public void markAsTemporarilyIdle() {
        this.output.markAsTemporarilyIdle();
    }

    @Override
    public Object getCheckpointLock() {
        return this.output.getCheckpointLock();
    }

    public void close() {
        this.output.close();
    }
}
